package com.cloud.point.concurrent.threadPool;

import com.cloud.point.concurrent.threadPool.TestThreadPool.TestBlockQueue;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StopWatch;

/**
 * 自定义模拟线程池实现
 *
 * @author 24373
 * @version 1.0
 * @date 2024-05-22 22:21
 */
@Slf4j
public class TestPool {
    public static void main(String[] args) {
        TestThreadPool threadPool = new TestThreadPool(1, 100, TimeUnit.MILLISECONDS, 5,
                (queue, task) -> {
                    task.run();
                }
        );
        StopWatch watch = new StopWatch();
        watch.start();
        for (int i = 0; i < 30; i++) {
            int j = i;
            //静态内部类不能使用外部类的非静态变量与方法
            threadPool.execute(() -> {
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                log.info("线程池执行任务" + j);
            });
        }
        watch.stop();
        log.info("任务执行完成，耗时：{}ms", watch.getTotalTimeMillis());
    }

}

@FunctionalInterface
interface TestRejectPolicy<T> {

    void reject(TestBlockQueue<T> queue, T task);

}

//@Slf4j
class TestThreadPool {
    private static final Logger log = LoggerFactory.getLogger(TestThreadPool.class);

    /**
     * 消息任务队列
     */
    private final TestBlockQueue<Runnable> blockQueue;
    /**
     * 消费者线程集合
     */
    private final HashSet<Worker> workers = new HashSet<Worker>();
    /**
     * 核心线程数
     */
    private final int coreSize;
    /**
     * 任务超时时间
     */
    private final long timeout;
    private final TimeUnit timeUnit;
    /**
     * 拒绝策略
     */
    private final TestRejectPolicy<Runnable> rejectPolicy;


    public TestThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapacity, TestRejectPolicy<Runnable> rejectPolicy) {

        this.coreSize = coreSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.blockQueue = new TestBlockQueue<Runnable>(queueCapacity);
        this.rejectPolicy = rejectPolicy;
    }

    public void execute(Runnable task) {
        synchronized (workers) {
            if (workers.size() < coreSize) {
                log.info("创建新worker");
                Worker worker = new Worker(task);
                workers.add(worker);
                //启动线程worker
                worker.start();
            } else {
                /**
                 * 队列添加策略
                 * 1.死等 2.超时等待 3.放弃执行 4.抛出异常 5.让调用者自己执行
                 *
                 * 分析：
                 * 一系列的方案太多，如果把每种方案都写在线程池中，代码会很复杂难以维护；可以封装成策略模式，抽象所有行为作为一个接口，让调用者自己决定使用哪种方案
                 */
                //blockQueue.offer(task, timeout, TimeUnit.MILLISECONDS);
                blockQueue.tryPut(rejectPolicy, task);
            }
        }
    }


    class Worker extends Thread {
        private Runnable task;

        public Worker(Runnable task) {
            this.task = task;
        }

        @Override
        public void run() {
            while (task != null || (task = blockQueue.poll(timeout, timeUnit)) != null) {
                try {
                    log.info("worker执行任务，task:{},worker:{}", task, this);
                    task.run();
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    log.info("worker执行任务执行完毕:{}", task);
                    task = null;
                }
            }
            synchronized (workers) {
                log.info("worker下线:{}", this);
                workers.remove(this);
            }
        }

    }

    public class TestBlockQueue<T> {

        /**
         * 1.任务队列
         */
        private Deque<T> queue = new ArrayDeque<T>();
        /**
         * 2.锁
         */
        private ReentrantLock lock = new ReentrantLock();
        /**
         * 3.生产者条件变量
         */
        private Condition fullCondition = lock.newCondition();
        /**
         * 4.消费者条件变量
         */
        private Condition emptyCondition = lock.newCondition();

        public TestBlockQueue(int capacity) {
            this.capacity = capacity;
        }

        /**
         * 5.队列容量
         */
        private int capacity;


        /**
         * 带超时的阻塞获取
         */
        public T poll(long timeout, TimeUnit unit) {
            lock.lock();
            try {
                long nanos = unit.toNanos(timeout);
                while (queue.isEmpty()) {
                    //    等待获取
                    try {
                        //超时没获取到，返回null
                        if (nanos <= 0) {
                            return null;
                        }
                        /**
                         * 避免虚假唤醒，重复等待相同时间
                         */
                        //awaitNanos返回的是剩余时间
                        nanos = emptyCondition.awaitNanos(nanos);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
                log.info("从阻塞队列获取到任务");
                T t = queue.removeFirst();
                //消费者消费完，唤醒生产者条件，添加线程
                fullCondition.signal();
                return t;
            } finally {
                lock.unlock();
            }
        }

        /**
         * 带拒绝策略的阻塞添加
         *
         * @param task
         */
        public void tryPut(TestRejectPolicy<T> rejectPolicy, T task) {
            lock.lock();
            try {
                if (queue.size() == capacity) {
                    rejectPolicy.reject(this, task);
                } else {
                    queue.addLast(task);
                    log.info("加入任务队列{}", task);
                    //队列有值了，唤醒消费者消费
                    emptyCondition.signal();
                }
            } finally {
                lock.unlock();
            }
        }

        /**
         * 带超时的阻塞添加
         *
         * @param element
         */
        public boolean offer(T element, long timeout, TimeUnit timeUnit) {
            lock.lock();
            try {
                long nanos = timeUnit.toNanos(timeout);
                while (queue.size() == capacity) {
                    // 队列满了；生产者条件阻塞，等待添加
                    try {
                        log.info("任务队列已满，等待加入任务队列{}", element);
                        if (nanos <= 0) {
                            return false;
                        }
                        nanos = fullCondition.awaitNanos(nanos);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
                queue.addLast(element);
                log.info("加入任务队列{}", element);
                //队列有值了，唤醒消费者消费
                emptyCondition.signal();
                return true;
            } finally {
                lock.unlock();
            }
        }

        /**
         * 阻塞获取
         */
        public T get() {
            lock.lock();
            try {
                while (queue.isEmpty()) {
                    //    等待获取
                    try {
                        emptyCondition.await();
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
                T t = queue.removeFirst();
                //消费者消费完，通知生产者生产
                fullCondition.signal();
                return t;
            } finally {

                lock.unlock();
            }
        }

        /**
         * 阻塞添加
         *
         * @param element
         */
        public void put(T element) {
            lock.lock();
            try {
                //todo 为什么不是 >=
                /**
                 * 精确性：只有当队列完全满时，才会触发拒绝策略。如果队列大小只是“大于或等于”容量，可能会在队列实际上还有空间时就错误地触发拒绝策略。
                 * 避免提前拒绝：如果使用 >=，在队列大小达到容量但尚未超出时（即等于的情况），就会触发拒绝。这实际上减少了队列的有效容量，因为它在队列还可以接受一个元素时就开始拒绝新元素。
                 * 最大化利用：使用 == 确保队列在达到其最大容量之前总是尝试接受新元素，从而最大化队列的使用效率。
                 */
                while (queue.size() == capacity) {
                    // 队列满了；生产者条件阻塞，等待添加
                    try {
                        log.info("任务队列已满，等待加入任务队列{}", element);
                        fullCondition.await();
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
                queue.addLast(element);
                log.info("加入任务队列{}", element);
                //队列有值了，唤醒消费者消费
                emptyCondition.signal();
            } finally {
                lock.unlock();
            }
        }

        public int size() {
            lock.lock();
            try {
                return queue.size();
            } finally {
                lock.unlock();
            }
        }
    }
}
